package bolts;

import java.util.Date;
import java.util.List;
import java.util.Map;

import output.TopWord;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.tuple.Tuple;

public class OutputTopWordsBolt extends AbstractOutputToFileBolt {

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        writer.println("Top Words at " + dateFormat.format(new Date()));

        final List<TopWord> topWords = (List<TopWord>) input.getValueByField("topWords");
        for (final TopWord word : topWords) {
            writer.println(word.getWord() + " " + word.getTweetCount() + " " + word.getFrequencyPerSecond());
        }
        writer.flush();
    }

    @Override
    protected String getOutputFileName(Map stormConfig) {
        return (String) stormConfig.get("topWordsOutputFile");
    }

}
